package com.shujia.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo21PageRank {
  def main(args: Array[String]): Unit = {


    /**
      *
      * 图存储的两种方式
      *
      * 1、存顶点
      * A   B,D
      * B   C
      * C   A,B
      * D   B,C
      *
      *
      *
      * 2、存边
      * A,B
      * B,C
      * A,D
      * C,A
      * C,B
      * D,C
      * D,B
      *
      */

    val conf: SparkConf = new SparkConf().setMaster("local[8]").setAppName("map")
    val sc: SparkContext = new SparkContext(conf)

    val prRDD: RDD[String] = sc.textFile("spark/data/pr.txt")


    var pageRankRDD: RDD[(String, List[String], Double)] = prRDD.map(line => {
      val split: Array[String] = line.split("\\|")

      //当前网页
      val page: String = split(0)

      //出链列表
      val link: List[String] = split(1).split(",").toList

      //初始pr值
      val pr: Double = split(2).toDouble
      (page, link, pr)
    })

    //网页数量
    val N: Long = pageRankRDD.count()


    val q: Double = 0.85

    var i: Double = 0.001
    var flag: Boolean = true

    while (flag) {

      val avgRDD: RDD[List[(String, Double)]] = pageRankRDD.map(line => {
        //当前网页
        val page: String = line._1

        //出链列表
        val link: List[String] = line._2

        //初始pr值
        val pr: Double = line._3

        //出链列表平分到的pr值
        val avgPr: Double = pr / link.length

        link.map(p => (p, avgPr))
      })

      //每个网页新的PR值
      val xinPR: RDD[(String, Double)] = avgRDD
        .flatMap(list => list)
        .reduceByKey(_ + _)
        .map(kv => {
          val page: String = kv._1
          val pr: Double = kv._2

          //增加阻尼系数
          val qPr: Double = pr * q + (1 - q) / N

          (page, qPr)
        })

      //关联出链列表
      val resultRDD: RDD[(String, List[String], Double)] = pageRankRDD
        .map(kv => (kv._1, kv._2))
        .join(xinPR)
        .map(kv => {
          val page: String = kv._1
          val link: List[String] = kv._2._1
          val pr: Double = kv._2._2
          (page, link, pr)
        })


      /**
        * 计算当前所有网页的pr值和上一次所有网页的pr的差值平均值
        *
        */

      val pageRankKVRDD: RDD[(String, Double)] = pageRankRDD.map(kv => (kv._1, kv._3))

      val joinRDD: RDD[(String, (Double, Double))] = resultRDD.map(kv => (kv._1, kv._3))
        .join(pageRankKVRDD)

      val chaRDD: RDD[Double] = joinRDD.map(kv => {
        val page: String = kv._1
        val currPr: Double = kv._2._1
        val lastPr: Double = kv._2._2
        math.abs(currPr - lastPr)
      })
      val chaAvg: Double = chaRDD.sum() / chaRDD.count()


      resultRDD.foreach(println)
      //如果差值的平均值小于0.001就收敛
      if (chaAvg < i) {
        flag = false
      }

      pageRankRDD = resultRDD

    }

    while (true){

    }
  }
}
